package cas.ihep.converter;

import alluxio.AlluxioURI;
import alluxio.client.file.FileSystem;
import alluxio.exception.AlluxioException;
import cas.ihep.alluxio.AlluxioBlock;
import cas.ihep.alluxio.AlluxioFile;
import cas.ihep.alluxio.AlluxioSystem;
import com.google.common.io.Closer;
import scala.Tuple2;

import java.io.*;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.file.StandardOpenOption;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.Scanner;

public class Root2Alluxio {
    public String basedir;

    public Root2Alluxio(){
        basedir="/tmp/"+Long.toHexString(System.currentTimeMillis())+'-'+Integer.toHexString(new Random().nextInt());
    }

    private Tuple2<List<String>,Long> processLocal0(BufferedReader reader,String filename,AlluxioSystem asys,String suffix)throws IOException{
        try(Scanner scanner=new Scanner(reader.readLine())) {
            long mcentries = scanner.nextLong();
            int filenum = scanner.nextInt();
            ArrayList<String> outlst = new ArrayList<>();
            filename = filename + suffix;
            for (int i = 0; scanner.hasNext(); i++) {
                String curfile = filename + i;
                String line = scanner.next();
                try (AlluxioFile cur = asys.create(new AlluxioURI(curfile))) {
                    AlluxioBlock blk = cur.nextBlock();
                    File dst = new File(blk.getBlockPath());
                    new File(line).renameTo(dst);
                    outlst.add(curfile);
                    blk.close();
                }
            }
            return new Tuple2<List<String>, Long>(outlst, mcentries);
        }
    }

    private Tuple2<List<String>,Long>[] processLocal(BufferedReader reader, String filename, AlluxioSystem asys)throws IOException{
        Tuple2<List<String>,Long>[] rets= new Tuple2[4];
        String decay=reader.readLine();
        try(AlluxioFile cur=asys.create(new AlluxioURI(filename+".decay"))){
            AlluxioBlock blk=cur.nextBlock();
            File dst=new File(blk.getBlockPath());
            new File(decay).renameTo(dst);
            ArrayList<String> tmp=new ArrayList<>();
            tmp.add(filename+".decay");
            blk.close();
            rets[0]=new Tuple2<List<String>,Long>(tmp, 0L);
        }
        rets[1]=processLocal0(reader,filename,asys,".mc.");
        rets[2]=processLocal0(reader,filename,asys,".bg.");
        rets[3]=processLocal0(reader,filename,asys,".dt.");
        return rets;
    }

    static class FileChannelInputStream extends InputStream{

        FileChannelInputStream(File file_,boolean deleteonclose)throws IOException{
            file=file_;
            channel=FileChannel.open(file.toPath(),StandardOpenOption.READ);
            buffer=channel.map(FileChannel.MapMode.READ_ONLY,0,file.length());
            autodelete=deleteonclose;
        }

        File file;
        FileChannel channel;
        MappedByteBuffer buffer;
        boolean autodelete;

        @Override
        public int read() {
            return buffer.get();
        }

        public int read(byte[] buf,int off,int len){
            int remain=buffer.remaining();
            if(remain<len){
                len=remain;
            }else if(remain==0){
                return -1;
            }
            buffer.get(buf,off,len);
            return len;
        }

        public int read(byte[] buf){
            int remain=buffer.remaining();
            if(remain>buf.length){
                remain=buf.length;
            }else if(remain==0){
                return -1;
            }
            buffer.get(buf,0,remain);
            return remain;
        }

        public int available(){
            return buffer.remaining();
        }

        public long skip(long sz){
            int cur=buffer.position();
            buffer.position((int) (cur+sz));
            return buffer.remaining();
        }

        @Override
        public void close() throws IOException {
            try {
                Method cleaner=buffer.getClass().getMethod("cleaner");
                cleaner.setAccessible(true);
                Object cleaner_object=cleaner.invoke(buffer);
                Method clean=cleaner_object.getClass().getMethod("clean");
                clean.setAccessible(true);
                clean.invoke(cleaner_object);
            }catch (NoSuchMethodException|IllegalAccessException | InvocationTargetException ex){
                throw new IOException(ex);
            }
            buffer=null;
            channel.close();
            if (autodelete){
                file.delete();
            }
        }
    }

    private Tuple2<List<String>,Long> processRemote0(BufferedReader reader,String filename,FileSystem afs,String suffix,byte[] buf)throws IOException,AlluxioException{
        try(Scanner scanner=new Scanner(reader.readLine())){
            long entries=scanner.nextLong();
            int filecount=scanner.nextInt();
            System.out.println("Reading entries "+entries+' '+filecount);
            List<String> outlst=new ArrayList<>();
            filename=filename+suffix;
            for(int i=0;scanner.hasNext();i++){
                String line=scanner.next();
                String curfile=filename+i;
                System.out.println("Reading from file "+line);
                File infile=new File(line);
                try(OutputStream os=afs.createFile(new AlluxioURI(curfile));
                    FileChannelInputStream is=new FileChannelInputStream(infile,true)){
                    int each;
                    while((each=is.read(buf))>0){
                        os.write(buf,0,each);
                    }
                    outlst.add(curfile);
                }
            }
            return new Tuple2<>(outlst,entries);
        }
    }

    private Tuple2<List<String>,Long>[] processRemote(BufferedReader reader, String filename, FileSystem afs) throws IOException, AlluxioException {
        Tuple2<List<String>,Long>[] rets= new Tuple2[4];
        byte[] buf=new byte[8192];
        String decay=reader.readLine();
        try(OutputStream os=afs.createFile(new AlluxioURI(filename+".decay"));
            FileChannelInputStream is=new FileChannelInputStream(new File(decay),true)){
            int each;
            while((each=is.read(buf))>0){
                os.write(buf,0,each);
            }
            List<String> tmp=new ArrayList<>();
            tmp.add(filename+".decay");
            rets[0]=new Tuple2<>(tmp,0L);
        }
        rets[1]=processRemote0(reader,filename,afs,".mc.",buf);
        rets[2]=processRemote0(reader,filename,afs,".bg.",buf);
        rets[3]=processRemote0(reader,filename,afs,".dt.",buf);
        return rets;
    }

    public Tuple2<List<String>,Long>[] convert(AlluxioSystem asys,String cmd,String jobname,int slices,List<String> mcs,List<String> bgs,List<String> dts) throws Exception {
        List<String> cmdlist=new ArrayList<>();
        cmdlist.add(cmd);
        //String filename=config.getString("filename");
        //cmdlist.add(config.getString("convertor")); cmdlist.add(filename); cmdlist.add(config.getString("treename"));
        //cmdlist.add(Integer.toString(config.getInt("events"))); cmdlist.add("20000"); cmdlist.addAll(config.getStringArray("branch"));
        ProcessBuilder builder=new ProcessBuilder();
        builder.redirectError(ProcessBuilder.Redirect.INHERIT);
        builder.command(cmdlist);
        Process subprocess=builder.start();
        Tuple2<List<String>,Long> ret[];
        jobname=basedir+'/'+jobname;
        try(Closer closer=Closer.create()){
            BufferedReader reader=closer.register(new BufferedReader(new InputStreamReader(subprocess.getInputStream())));
            PrintWriter writer=closer.register(new PrintWriter(subprocess.getOutputStream()));
            writer.print(slices); writer.print(' '); writer.print(mcs.size()); writer.print(' ');
            writer.print(bgs.size()); writer.print(' '); writer.println(dts.size());
            for(String s:mcs){
                writer.println(s);
            }
            for(String s:bgs){
                writer.println(s);
            }
            for(String s:dts){
                writer.println(s);
            }
            writer.flush();
            if (asys.hasLocalWorker()){
                ret=processLocal(reader,jobname,asys);
            }else{
                ret=processRemote(reader,jobname,asys.getFileSystem());
            }
            if (subprocess.waitFor()!=0){
                throw new RuntimeException("Exit error");
            }
        }
        subprocess.destroy();
        return ret;
    }
}
